package com.xiashitech.agent.grpc;

import com.xiashitech.agent.config.StartConfig;
// import com.xiashitech.agent.service.extload.AuthService;
import com.xiashitech.agent.db.extload.LocalDataSource;
import com.xiashitech.agent.service.extloadproxy.AuthServiceProxy;
import com.xiashitech.agent.utils.JsonUtil;
import com.xiashitech.interfaces.grpc.RPCIncrementalCacheRequest;
import com.xiashitech.interfaces.grpc.RPCIncrementalCacheResponse;
import com.xiashitech.interfaces.grpc.RPCIncrementalCacheServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.stub.StreamObserver;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class XsIncrementClient {
    private final static XsIncrementClient client = new XsIncrementClient();
    private RPCIncrementalCacheServiceGrpc.RPCIncrementalCacheServiceStub requestStub = null;
    private StreamObserver<RPCIncrementalCacheRequest> requestStreamObserver;
    private volatile boolean isError = false;
    private volatile boolean isReReg = false;
    private String regInfoSaved = "";
    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
    public static XsIncrementClient getClient() {
        return client;
    }

    public void regIncrementCacheObserver(String regInfo) {
        System.out.println(Thread.currentThread().getId() + ", regIncrementCacheObserver enter, StartConfig.agentService: " + StartConfig.getParamValueFromKey(StartConfig.agentService));
        ManagedChannel channelRef = XsChannelManager.getManager().buildStubOfChannel("192.168.0.54", 9988);
        System.out.println(Thread.currentThread().getId() + ", regIncrementCacheObserver, now channel identityHashCode: " + System.identityHashCode(channelRef));
        synchronized (client) {
            if(requestStub == null) {
                final StreamObserver<RPCIncrementalCacheResponse> requestSO = new StreamObserver<RPCIncrementalCacheResponse>() {
                    @Override
                    public void onNext(RPCIncrementalCacheResponse value) {
                        String respStr = value.getRespJson();
                        if("INCREMENTHEARTBEAT".equals(respStr)) // 用于维持 nginx 代理的心跳
                            return;
                        System.out.println(Thread.currentThread().getId() + ", regIncrementCacheObserver, receive the server pushed content: " + value);
                        // AuthService.getInstance().onMessage(respStr);
                        AuthServiceProxy.onMessage(respStr);
                    }

                    @Override
                    public void onError(Throwable t) {
                        System.out.println(Thread.currentThread().getId() + ", regIncrementCacheObserver: error: " + t.getMessage());
                        isError = true;
                    }

                    @Override
                    public void onCompleted() {
                        onCompleted();
                    }
                };
                long period = 55; // 55 秒
                if(!StartConfig.isProxyGrpc()) {
                    period = 600; // 10 分钟
                }
                scheduledExecutorService.scheduleAtFixedRate(() -> {
                    try {
                        Thread.currentThread().setName("XsIncrementHeartBeat");
                        if(requestStub == null) {
                            System.out.println(Thread.currentThread().getId() + " T" + System.nanoTime() + ", XsIncrementHeartBeat requestStub is NULL .......... SHOULDNOT exist");
                            requestStub = RPCIncrementalCacheServiceGrpc.newStub(channelRef);
                        }
                        if(isError || requestStreamObserver == null) {
                            requestStreamObserver = requestStub.biStreamIncrementalCache(requestSO);
                            isError = false;
                            isReReg = true;
                        }
                        if(requestStreamObserver != null ) {
                            System.out.println(Thread.currentThread().getId() + " T" + System.nanoTime() + ", request a Increment heart beart");
                            RPCIncrementalCacheRequest request = RPCIncrementalCacheRequest.newBuilder().setReqJson("INCREMENTHEARTBEAT").build();
                            if(isReReg) {
                                if(regInfoSaved == null || regInfoSaved.isEmpty()) {
                                    String systemId = StartConfig.getParamValueFromKey(StartConfig.systemId);
                                    Map<String, Object> paramMap = new HashMap<>();
                                    paramMap.put("systemId", systemId);
                                    paramMap.put("serviceId", System.getProperty("otel.service.name", systemId));
                                    paramMap.put("serialId", LocalDataSource.agentInstanceId);
                                    // 追加一个求serialId Hash的原始字符串（调试过程方便查看相关原始信息）
                                    paramMap.put("serialIdOrigin", LocalDataSource.agentInstanceIdOrigin);
                                    regInfoSaved = JsonUtil.convertObjectToString(paramMap);
                                }
                                request = RPCIncrementalCacheRequest.newBuilder().setReqJson(regInfoSaved).build();
                                isReReg = false;
                            }
                            synchronized (requestStreamObserver) { // 应该不会存在对 soAgentHeartBeat 的并发调用
                                System.out.println(Thread.currentThread().getId() + " T" + System.nanoTime() + ", XsIncrementClient onNext######### requestStreamObserver: " + System.identityHashCode(requestStreamObserver));
                                requestStreamObserver.onNext(request);
                            }
                        } else { // 该变量为空，需要重连，会和链接断掉重连冲突吗？应该不会，在 agentStatus 中有锁，不会多次创建 soAgentHeartBeat
                            System.out.println(Thread.currentThread().getId() + " T" + System.nanoTime() + ", XsIncrementClient requestStreamObserver is null !!! SHOULDNOT");
                        }
                    } catch (Throwable t) {
                        StartConfig.throwableStackPrinter(t);
                    }
                }, 2, period, TimeUnit.SECONDS);

                // 初始化 stub 和 StreamObserver
                requestStub = RPCIncrementalCacheServiceGrpc.newStub(channelRef);
                requestStreamObserver = requestStub.biStreamIncrementalCache(requestSO);
            }
        }
        System.out.println(Thread.currentThread().getId() + " T" + System.nanoTime() + ", XsIncrementClient regIncrementCacheObserver (OnlyOne) regInfo: " + regInfo);
        RPCIncrementalCacheRequest request = RPCIncrementalCacheRequest.newBuilder().setReqJson(regInfo).build();
        synchronized (requestStreamObserver) {
            requestStreamObserver.onNext(request);
            regInfoSaved = regInfo;
        }
    }
}
